package org.example

import org.apache.spark.sql.{DataFrame, SparkSession}

object ElasticsearchSparkSQLExample {

  def main(args: Array[String]): Unit = {
    // 创建SparkSession
    val spark = SparkSession
      .builder()
      .appName("ElasticsearchSparkSQLExample")
      .config("spark.es.nodes", "localhost") // Elasticsearch节点地址
      .config("spark.es.port", "9200") // Elasticsearch端口
      .master("local[*]")
      .getOrCreate()

    //读取Elasticsearch数据为DataFrame=====================================================================================================
//    val df: DataFrame = spark.read
//      .format("org.elasticsearch.spark.sql")
//      .option("es.index.auto.create", "false") // 不自动创建索引
//      .option("es.resource", "user/_doc") // 索引名和类型名（在Elasticsearch 7.x+中，类型通常是_doc）
//      .load()
    //读取json数据为DataFrame=====================================================================================================
    val df = spark.read.json("data/users.json")

    //下面是通用操作 ====================================================================================================

    // 显示DataFrame的schema
    df.printSchema()
    // 显示前几条数据
    df.printSchema()

    // 使用SQL语句查询数据
    df.createOrReplaceTempView("user")
    val sqlDF = spark.sql("SELECT userName, address FROM user WHERE useId = 12")

    // 显示查询结果
    sqlDF.show(10, truncate = false)

    // 停止SparkSession
    spark.stop()
  }
}
